package belf.migrate.plugin.oracle;

import belf.migrate.api.common.DatabaseIdentifier;
import belf.migrate.api.constants.SQLConstants;
import belf.migrate.api.exception.*;

import belf.migrate.api.model.TableInfoModel;
import belf.migrate.api.table.schema.*;
import belf.migrate.api.taskconf.TaskConf;
import belf.migrate.api.util.CatalogUtils;
import belf.migrate.api.util.ExceptionUtils;
import belf.migrate.api.util.JdbcUrlUtil;
import belf.migrate.core.util.StringTool;
import belf.migrate.plugin.oracle.antlr.PlSqlLexer;
import belf.migrate.plugin.oracle.antlr.PlSqlParser;
import belf.migrate.plugin.oracle.listener.Oracle2DMFunctionListener;

import lombok.extern.slf4j.Slf4j;

import org.antlr.v4.runtime.CharStream;
import org.antlr.v4.runtime.CharStreams;
import org.antlr.v4.runtime.CommonTokenStream;
import org.antlr.v4.runtime.tree.ParseTree;
import org.antlr.v4.runtime.tree.ParseTreeWalker;
import org.apache.commons.lang3.StringUtils;

import java.sql.*;
import java.util.*;
import java.util.function.Predicate;

@Slf4j
public class OracleCatalog extends AbstractJdbcCatalog {
    private static final String QUOTE = "\"";

    public OracleCatalog(String username, String pwd, JdbcUrlUtil.UrlInfo urlInfo, String defaultSchema, boolean caseSensitive) {
        super(DatabaseIdentifier.ORACLE, username, pwd, urlInfo, defaultSchema, caseSensitive);
    }

    @Override
    public String getFullTableName(TablePath tablePath) {
        return tablePath.getSchemaAndTableName();
    }

    @Override
    public String getQuote() {
        return QUOTE;
    }

    @Override
    public String getListTableSQL(String database, String schema) {
        return String.format("""
                SELECT OBJECT_NAME AS TABLE_NAME
                FROM all_objects
                WHERE OBJECT_TYPE='TABLE' AND OWNER='%s'""",
                schema.toUpperCase());
    }

    @Override
    public String getListDatabaseSQL() {
        //FIXME: 获取当前用户名,需要决定是获取所有有权限的schema还是当前schema
//        return "SELECT SYS_CONTEXT('USERENV','CURRENT_SCHEMA') AS CURRENT_SCHEMA FROM DUAL";
        return "SELECT distinct owner FROM all_tables";
    }

    /**
     * oracle查询指定schema下的所有视图，实际是查询用户空间的所有视图
     * @param schemaName
     * @return
     */
    @Override
    public String getListViewsSQL(String schemaName) {
        return "SELECT VIEW_NAME \n" +
                "FROM ALL_VIEWS \n" +
                "WHERE OWNER = '" + schemaName + "'";
    }

    @Override
    public String getListViewsSQLInfo(String defaultDatabase, String schemaName) {
        return "SELECT VIEW_NAME \n" +
                "FROM ALL_VIEWS \n" +
                "WHERE OWNER = '" + schemaName + "'";
    }

    @Override
    public String getListFunctionSQL(String schemaName) {
        return String.format("SELECT OBJECT_NAME AS FUNCTION_NAME FROM ALL_OBJECTS WHERE OBJECT_TYPE = 'FUNCTION' AND OWNER = '%s'", schemaName);
    }

    @Override
    public String getListTriggerSQL(String schemaName) {
        return String.format("SELECT TRIGGER_NAME \n" +
                "FROM ALL_TRIGGERS \n" +
                "WHERE OWNER = '%s'", schemaName);
    }

    @Override
    public String getListIndexSQL(String schemaName) {
        return String.format("SELECT i.INDEX_NAME,i.TABLE_NAME FROM ALL_INDEXES i WHERE i.OWNER='%s' ORDER BY i.INDEX_NAME", schemaName);
    }

    @Override
    public String getListTypeSQL(String defaultDatabase) {
        return String.format("SELECT TYPE_NAME FROM ALL_TYPES WHERE OWNER='%s' ORDER BY TYPE_NAME", defaultDatabase);
    }

    @Override
    public String getListProcedureSQL(String defaultDatabase) {
        return String.format("SELECT OBJECT_NAME AS PROCEDURE_NAME " +
                "FROM ALL_OBJECTS " +
                "WHERE OBJECT_TYPE = 'PROCEDURE' " +
                "AND OWNER='%s' ORDER BY OBJECT_NAME", defaultDatabase);
    }

    @Override
    public String getExistDataSql(TablePath tablePath) {
        return "SELECT COUNT(*) FROM " + tablePath.getTableName();
    }

    @Override
    public String getCountDataSql(TablePath tablePath) {
        return "SELECT COUNT(*) FROM " + tablePath.getSchemaAndTableName();
    }

    @Override
    public List<TableInfoModel> listTablesInfo(String databaseName, String schemaName)
            throws CatalogException, DatabaseNotExistException {

        try (PreparedStatement ps =
                     getConnection(defaultUrl)
                             .prepareStatement(getListTableSQL(databaseName, schemaName));
             ResultSet rs = ps.executeQuery()) {

            List<TableInfoModel> tables = new ArrayList<>();
            while (rs.next()) {
//                String fullTableName = String.format("\"%s\".\"%s\"",
//                        schemaName,
//                        rs.getString(1));
                TableInfoModel table = new TableInfoModel();
                table.setTableName(rs.getString("TABLE_NAME"));
//                table.setRows(rs.getLong("TABLE_ROWS"));
//                table.setComment(rs.getString("TABLE_COMMENT"));
                tables.add(table);
            }

            return tables;
        } catch (Exception e) {
            throw new CatalogException(
                    String.format("Failed listing table in catalog %s", catalogName), e);
        }
    }

    @Override
    public TableInfoModel getTableInfoByName(TablePath tablePath) throws CatalogException {
        TableInfoModel table = new TableInfoModel();
        try {
            String sql = "SELECT\n" +
                    "\tut.table_name AS TABLE_NAME,\n" +
                    "\tut.num_rows AS TABLE_ROWS,\n" +
                    "\tROUND(us.bytes / (1024 * 1024), 2) AS TABLE_SIZE\n" +
                    "FROM\n" +
                    "\tUSER_TABLES ut\n" +
                    "JOIN USER_SEGMENTS us ON\n" +
                    "\tut.table_name = us.segment_name\n" +
                    "\tAND us.segment_type = 'TABLE' AND ut.table_name = ?";
            log.info("Query mysql table info: {}", sql);
            PreparedStatement preparedStatement = getConnection(defaultUrl).prepareStatement(sql);
            preparedStatement.setString(1, tablePath.getTableName());
            ResultSet resultSet = preparedStatement.executeQuery();
            if (resultSet.next()) {
                table.setTableName(resultSet.getString("TABLE_NAME"));
                table.setRows(resultSet.getLong("TABLE_ROWS"));
                table.setSize(resultSet.getDouble("TABLE_SIZE"));
            }
            resultSet.close();
            preparedStatement.close();
        } catch (SQLException e) {
            log.error("获取表的信息失败！", e);
        }
        return table;
    }

    @Override
    public String getListSequenceSQL(String schema) {
        return String.format("SELECT SEQUENCE_NAME FROM ALL_SEQUENCES WHERE SEQUENCE_OWNER = '%s'", schema);
    }

    @Override
    public String getListSynonymSQL(String schema) {
        return String.format("SELECT SYNONYM_NAME FROM ALL_SYNONYMS WHERE OWNER='%s'", schema);
    }

    @Override
    protected void fillingColumnExtra(TablePath tablePath, List<ColumnMetaData> columns, Connection connection) {
        String sql = """
                SELECT 
                    column_name AS COLUMN_NAME, data_default AS COLUMN_DEFAULT 
                FROM 
                    ALL_TAB_COLUMNS
                WHERE 
                    owner = ? 
                    AND table_name = ?
                """;
        try (PreparedStatement ps = connection.prepareStatement(sql)) {
            ps.setString(1, tablePath.getSchemaName());
            ps.setString(2, tablePath.getTableName());

            ResultSet rs = ps.executeQuery();
            while (rs.next()) {
                String columnName = rs.getString("COLUMN_NAME");
                String columnDefault = rs.getString("COLUMN_DEFAULT");

                Optional<ColumnMetaData> possibleColumn = columns.stream().filter(new Predicate<ColumnMetaData>() {
                    @Override
                    public boolean test(ColumnMetaData columnMetaData) {
                        return columnMetaData.getColumnName().equalsIgnoreCase(columnName);
                    }
                }).findFirst();
                if (possibleColumn.isPresent()) {
                    ColumnMetaData columnMetaData = possibleColumn.get();
                    ColumnExtra columnExtra = new ColumnExtra(columnDefault, null, null, 0);
                    columnMetaData.setExtra(columnExtra);
                }
            }
        } catch (SQLException e) {
            log.error("Error on query oracle column extra information:\n{}", ExceptionUtils.toString(e));
        }
    }

    @Override
    public List<IndexKey> getIndexKeys(TablePath tablePath) throws TableNotExistException, CatalogException {
        if (!tableExists(tablePath)) {
            throw new TableNotExistException(catalogName, tablePath);
        }

        String dbUrl = defaultUrl;

        // 查询指定表的索引，排除主键索引
        String query = "SELECT \n" +
                "    ind.INDEX_NAME AS INDEX_NAME, \n" +
                "    ind.ITYP_NAME AS INDEX_TYPE_NAME,\n" +
                "    ind.UNIQUENESS AS UNIQUENESS," +
                "    col.COLUMN_NAME AS COLUMN_NAME, \n" +
                "    col.COLUMN_POSITION AS SEQ_IN_INDEX,\n" +
                "    col.DESCEND AS INDEX_ORDER\n" +
                "FROM \n" +
                "    ALL_INDEXES ind\n" +
                "JOIN \n" +
                "    ALL_IND_COLUMNS col \n" +
                "ON \n" +
                "    ind.INDEX_NAME = col.INDEX_NAME\n" +
                "WHERE \n" +
                "    ind.TABLE_OWNER = '%s' \n" +
                "    AND ind.TABLE_NAME = '%s' \n" +
                "    AND ind.INDEX_TYPE NOT LIKE 'FUNCTION-BASED NORMAL'" +  //排除函数类索引
                "    AND ind.INDEX_NAME NOT IN (\n" +
                "        SELECT cc.constraint_NAME\n" +
                "        FROM ALL_CONSTRAINTS c\n" +
                "        JOIN ALL_CONS_COLUMNS cc\n" +
                "        ON c.CONSTRAINT_NAME = cc.CONSTRAINT_NAME\n" +
                "        WHERE c.CONSTRAINT_TYPE = 'P'\n" +
                "        AND c.TABLE_NAME = '%s'\n" +
                "        AND c.OWNER = '%s'\n" +
                "    )\n" +
                "ORDER BY \n" +
                "    ind.INDEX_NAME, col.COLUMN_POSITION";

        try {
            Connection conn = getConnection(dbUrl);
            String sql = String.format(query,
                    tablePath.getSchemaName(),
                    tablePath.getTableName(),
                    tablePath.getTableName(),
                    tablePath.getSchemaName());
            Statement statement = conn.createStatement();

            DatabaseMetaData metaData = conn.getMetaData();
            List<IndexKey> defaultIndexKeys =  CatalogUtils.getIndexKeys(metaData, tablePath);

            // index name -> index
            Map<String, IndexKeyExt> indexKeyMap = new HashMap<>();

            try (ResultSet rs = statement.executeQuery(sql)) {
                log.debug("query index:{}", sql);
                while (rs.next()) {
                    boolean unique = rs.getString("UNIQUENESS").equalsIgnoreCase("NONUNIQUE")?false:true;
                    String indexName = rs.getString("INDEX_NAME");
                    int sequenceInIndex = rs.getInt("SEQ_IN_INDEX");
                    String columnName = rs.getString("COLUMN_NAME");
                    String indexOrder = rs.getString("INDEX_ORDER");
                    String indexTypeName = rs.getString("INDEX_TYPE_NAME");
                    List<IndexKeyExt.ColumnWithSequence> columns = new ArrayList<>();
                    columns.add(new IndexKeyExt.ColumnWithSequence(columnName, sequenceInIndex));
                    IndexKeyExt IndexKeyExt = new IndexKeyExt(
                            columns,
                            indexName,
                            unique,
                            indexOrder,
                            indexTypeName);
                    IndexKeyExt existOne = indexKeyMap.get(indexName);
                    if (existOne != null) {
                        existOne.getColumns().addAll(IndexKeyExt.getColumns());
                    } else {
                        indexKeyMap.put(indexName, IndexKeyExt);
                    }

                    log.debug("got index: {}", IndexKeyExt);
                }
            }

            Iterator<IndexKey> it = defaultIndexKeys.iterator();
            while (it.hasNext()) {
                IndexKey indexKey = it.next();
                IndexKeyExt IndexKeyExt = indexKeyMap.get(indexKey.getIndexName());
                if (null != IndexKeyExt) {
                    indexKey.setColumnNames(IndexKeyExt.getSortedColumnNames());
                    indexKey.setIndexTypeName(IndexKeyExt.getIndexTypeName());
                    if (IndexKeyExt.getIndexOrder() != null
                            && IndexKeyExt.getIndexOrder().equalsIgnoreCase("ASC")) {
                        indexKey.setIndexOrder(IndexKey.IndexOrder.ASC);
                    } else if (IndexKeyExt.getIndexOrder() != null
                            && IndexKeyExt.getIndexOrder().equalsIgnoreCase("DESC")) {
                        indexKey.setIndexOrder(IndexKey.IndexOrder.DESC);
                    }
                } else { // oracle自身查找索引不包含主键和Function-based Normal索引，把这些索引相应删除了
                    it.remove();
                }
            }

            statement.close();
            return defaultIndexKeys;
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 显示Oracle创建表的语句。
     * @param tablePath
     * @return
     * @throws ViewNotExistException
     */
    @Override
    public String showCreateTable(TablePath tablePath) throws TableNotExistException {
        String query = "SELECT DBMS_METADATA.GET_DDL('TABLE', '%s', '%s') AS TABLE_DDL FROM DUAL";
        String sql = String.format(query, tablePath.getTableName(), tablePath.getSchemaName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("TABLE_DDL");
            }
        } catch (SQLException e) {
            log.error("Error while show create view: {}", sql);
            e.printStackTrace();
            throw new TableNotExistException(DatabaseIdentifier.ORACLE, tablePath);
        }
        return null;
    }

    /**
     * 显示Oracle创建视图的语句。
     * @param tablePath
     * @return
     * @throws ViewNotExistException
     */
    @Override
    public String showCreateView(TablePath tablePath) throws ViewNotExistException {
        String query = "SELECT DBMS_METADATA.GET_DDL('VIEW', '%s', '%s') AS view_ddl FROM DUAL";
        String sql = String.format(query, tablePath.getTableName(), tablePath.getSchemaName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("VIEW_DDL");
            }
        } catch (SQLException e) {
            log.error("Error while show create view: {}", sql);
            e.printStackTrace();
            throw new ViewNotExistException(DatabaseIdentifier.ORACLE, tablePath);
        }
        return null;
    }

    @Override
    public String showCreateProcedure(TablePath tablePath) throws ProcedureNotExistException {
        String query = "SELECT DBMS_METADATA.GET_DDL('PROCEDURE', '%s', '%s') AS PROCEDURE_DDL FROM DUAL";
        String sql = String.format(query, tablePath.getTableName(), tablePath.getSchemaName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("PROCEDURE_DDL");
            }
        } catch (SQLException e) {
            log.error("Error while show create procedure : \n{}", sql);
            throw new ProcedureNotExistException(DatabaseIdentifier.ORACLE, tablePath);
        }
        return null;
    }

    @Override
    public String showCreateTrigger(TablePath tablePath) throws TriggerNotExistException{
        String query = "SELECT DBMS_METADATA.GET_DDL('TRIGGER', '%s', '%s') AS trigger_ddl FROM DUAL";
        String sql = String.format(query, tablePath.getTableName(), tablePath.getSchemaName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("TRIGGER_DDL");
            }
        } catch (SQLException e) {
            log.error("Error while show create trigger: \n{}", sql);
            throw new TriggerNotExistException(DatabaseIdentifier.ORACLE, tablePath, e);
        }
        return null;
    }

    @Override
    public String showCreateType(TablePath tablePath) throws TypeNotExistException {
        String query = "SELECT DBMS_METADATA.GET_DDL('TYPE', '%s', '%s') AS type_ddl FROM DUAL";
        String sql = String.format(query, tablePath.getTableName(), tablePath.getSchemaName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("TYPE_DDL");
            }
        } catch (SQLException e) {
            log.error("Error while show create type: \n{}", sql);
            throw new TypeNotExistException(DatabaseIdentifier.ORACLE, tablePath, e);
        }
        return null;
    }

    @Override
    public String showCreateSeq(TablePath tablePath) throws SequenceNotExistException {
        String query = "SELECT DBMS_METADATA.GET_DDL('SEQUENCE', '%s', '%s') AS seq_ddl FROM DUAL";
        String sql = String.format(query, tablePath.getTableName(), tablePath.getSchemaName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("SEQ_DDL");
            }
        } catch (SQLException e) {
            log.error("Error while show create sequence: \n{}", sql);
            throw new SequenceNotExistException(DatabaseIdentifier.ORACLE, tablePath, e);
        }
        return null;
    }

    @Override
    public String showCreateSynonym(TablePath tablePath) throws SequenceNotExistException {
        String query = "SELECT DBMS_METADATA.GET_DDL('SYNONYM', '%s', '%s') AS synonym_ddl FROM DUAL";
        String sql = String.format(query, tablePath.getTableName(), tablePath.getSchemaName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("SYNONYM_DDL");
            }
        } catch (SQLException e) {
            log.error("Error while show create sequence: \n{}", sql);
            throw new SequenceNotExistException(DatabaseIdentifier.ORACLE, tablePath, e);
        }
        return null;
    }

    @Override
    public String showCreateFunc(TablePath tablePath) throws FunctionNotExistException {
        String query = "SELECT DBMS_METADATA.GET_DDL('FUNCTION', '%s', '%s') AS func_ddl FROM DUAL";
        String sql = String.format(query, tablePath.getTableName(), tablePath.getSchemaName());
        try {
            Statement statement = getConnection(defaultUrl).createStatement();
            ResultSet resultSet = statement.executeQuery(sql);
            if (resultSet.next()) {
                return resultSet.getString("FUNC_DDL");
            }
        } catch (SQLException e) {
            log.error("Error while show create function: \n{}", sql);
            throw new FunctionNotExistException(DatabaseIdentifier.ORACLE, tablePath, e);
        }
        return null;
    }

    @Override
    public String showCreateIndex(TablePath tablePath) throws IndexNotExistException {
        String query = "SELECT DBMS_METADATA.GET_DDL('INDEX', '%s', '%s') AS INDEX_DDL FROM DUAL";
        String sql = String.format(query, tablePath.getTableName(), tablePath.getSchemaName());
        try {
            return getDDL(sql, "INDEX_DDL");
        } catch (CatalogException e) {
            log.error("Error while show create index: \n{}", sql);
            throw new IndexNotExistException(DatabaseIdentifier.ORACLE, tablePath, e);
        }
    }

    private final static String QUERY_UNIQUE = """
            SELECT
                a.constraint_name AS CONSTRAINT_NAME,
                b.column_name AS COLUMN_NAME
            FROM
                user_constraints a
                JOIN user_cons_columns b
                ON a.constraint_name = b.constraint_name
            WHERE
            	a.OWNER = ?
            	AND a.TABLE_NAME = ?
            	AND a.CONSTRAINT_TYPE = 'U'
            """;
    @Override
    public List<UniqueKey> getUniqueKeys(TablePath tablePath) throws CatalogException {
        Map<String, UniqueKey> uniqueKeyMap = new HashMap<>();
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);

        try(PreparedStatement ps = connection.prepareStatement(QUERY_UNIQUE)) {
            ps.setString(1, tablePath.getSchemaName());
            ps.setString(2, tablePath.getTableName());
            try (ResultSet rs = ps.executeQuery()){

                while (rs.next()) {
                    String constraintName = rs.getString("CONSTRAINT_NAME");
                    String column = rs.getString("COLUMN_NAME");
                    if (uniqueKeyMap.containsKey(constraintName)) {
                        UniqueKey value = uniqueKeyMap.get(constraintName);
                        value.addColumn(column);
                    } else {
                        List<String> columns = new ArrayList<>();
                        columns.add(column);
                        UniqueKey uniqueKey = new UniqueKey(tablePath.getTableName(), constraintName, columns);
                        uniqueKeyMap.put(constraintName, uniqueKey);
                    }
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, "Oracle查询Unique约束失败：table=" + tablePath.getTableName(), e);
        }
        return uniqueKeyMap.values().parallelStream().toList();
    }

    private final static String QUERY_CHECK_CONSTRAINT = """
            SELECT
                a.constraint_name AS CONSTRAINT_NAME,
                a.SEARCH_CONDITION AS SEARCH_CONDITION
            FROM
                user_constraints a
            WHERE
            	a.OWNER = ?
            	AND a.TABLE_NAME = ?
            	AND a.CONSTRAINT_TYPE = 'C'
            """;
    @Override
    public List<CheckConstraint> getCheckConstraints(TablePath tablePath) throws CatalogException {
        List<CheckConstraint> checkConstraints = new ArrayList<>();
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);

        try(PreparedStatement ps = connection.prepareStatement(QUERY_CHECK_CONSTRAINT)) {
            ps.setString(1, tablePath.getSchemaName());
            ps.setString(2, tablePath.getTableName());
            try (ResultSet rs = ps.executeQuery()){
                while (rs.next()) {
                    String constraintName = rs.getString("CONSTRAINT_NAME");
                    String condition = rs.getString("SEARCH_CONDITION");
                    if (!(condition.toUpperCase().contains("NOT NULL"))) { //对于oracle字段非空的约束，已经在字段构建里处理了，无需再单独约束
                        checkConstraints.add(new CheckConstraint(tablePath.getTableName(),
                                constraintName,
                                condition));
                    }
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, "查询Check约束失败：table=" + tablePath.getTableName(), e);
        }
        return checkConstraints;
    }

    @Override
    public List<UserDefType> listTypes(String schema, List<String> typeNames) throws CatalogException {
        String sql = "SELECT TYPE_NAME,TYPECODE,SUPERTYPE_NAME  FROM user_types";
        if (null != typeNames && !typeNames.isEmpty()) {
            List<String> quotedTypeNames = new ArrayList<>();
            for (String typeName: typeNames) {
                quotedTypeNames.add("'" + typeName + "'");
            }
            sql += String.format(" WHERE TYPE_NAME IN (%s)", String.join(",", quotedTypeNames));
        }
        List<UserDefType> userDefTypes = new ArrayList<>();
        String dbUrl = defaultUrl;
        Connection connection = getConnection(dbUrl);

        try (Statement stmt = connection.createStatement()){
            try (ResultSet rs = stmt.executeQuery(sql)){
                while (rs.next()) {
                    UserDefType userDefType = new UserDefType();
                    String typeName = rs.getString("TYPE_NAME");
                    userDefType.setTypeName(typeName);
                    userDefType.setTypeCode(rs.getString("TYPECODE"));
                    userDefType.setSuperTypeName(rs.getString("SUPERTYPE_NAME"));

                    String ddlQuery = String.format("SELECT DBMS_METADATA.GET_DDL('%s', '%s', '%s') AS DDL FROM dual",
                            "TYPE",
                            typeName,
                            schema);
                    try (PreparedStatement ps = connection.prepareStatement(ddlQuery)){
                        ResultSet ddlResultSet = ps.executeQuery();
                        if (ddlResultSet.next()) {
                            userDefType.setDdl(ddlResultSet.getString("DDL"));
                        }
                    } catch (SQLException e) {
                        e.printStackTrace();
                        throw new CatalogException(APIErrorCode.QUERY_TYPE_DDL_ERROR, "查询自定义TYPE的DDL失败：TYPE=" + typeName, e);
                    }
                    userDefTypes.add(userDefType);
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.QUERY_TYPE_LIST_ERROR, "查询所有自定义TYPE的名称失败，请检查SQL语句：sql=" + sql, e);
        }
        return userDefTypes;
    }

    public List<UserDefSeq> listSequences(String owner, List<String> sequenceNames) throws CatalogException {

        List<UserDefSeq> userDefSeqs = new ArrayList<>();
        String dbUrl = defaultUrl;
        Connection connection = getConnection(dbUrl);
        Map<String, String> seqDDLs = new OracleDDLBuilder().buildSequenceDDL(connection, owner, sequenceNames);
        for (Map.Entry<String, String> entry : seqDDLs.entrySet()) {
            userDefSeqs.add(new UserDefSeq(entry.getKey(), entry.getValue()));
        }
        return userDefSeqs;
    }

    public List<UserDefFunction> listFunctions(String schema, List<String> functionNames) throws CatalogException {
        String sql = "SELECT OBJECT_NAME FROM ALL_OBJECTS WHERE OBJECT_TYPE = 'FUNCTION' AND OWNER = '" + schema +"'";
        if (null != functionNames && !functionNames.isEmpty()) {
            List<String> quotedFuncNames = new ArrayList<>();
            for (String funcName: functionNames) {
                quotedFuncNames.add("'" + funcName + "'");
            }
            sql += String.format(" AND OBJECT_NAME IN (%s)", String.join(",", quotedFuncNames));
        }
        List<UserDefFunction> userDefFunctions = new ArrayList<>();
        String dbUrl = defaultUrl;
        Connection connection = getConnection(dbUrl);

        try (Statement stmt = connection.createStatement()){
            try (ResultSet rs = stmt.executeQuery(sql)){
                while (rs.next()) {
                    UserDefFunction userDefFunction = new UserDefFunction();
                    String functionName = rs.getString("OBJECT_NAME");
                    userDefFunction.setFunctionName(functionName);
                    String ddlQuery = String.format("SELECT DBMS_METADATA.GET_DDL('%s', '%s', '%s') AS DDL FROM dual",
                            "FUNCTION",
                            functionName,
                            schema);
                    try (PreparedStatement ps = connection.prepareStatement(ddlQuery)){
                        ResultSet ddlResultSet = ps.executeQuery();
                        if (ddlResultSet.next()) {
                            userDefFunction.setDdl(ddlResultSet.getString("DDL"));
                        }
                    } catch (SQLException e) {
                        e.printStackTrace();
                        throw new CatalogException(APIErrorCode.QUERY_FUNCTION_DDL_ERROR, "查询自定义函数的DDL失败：function=" + functionName, e);
                    }
                    userDefFunctions.add(userDefFunction);
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.QUERY_FUNCTION_LIST_ERROR, "查询所有自定义函数的DDL失败，请检查SQL语句：sql=" + sql, e);
        }
        return userDefFunctions;
    }


    public List<UserDefSynonym> listSynonyms(String schema, List<String> synonymNames) throws CatalogException {
        String sql = "SELECT SYNONYM_NAME FROM ALL_SYNONYMS";
        if (null != synonymNames && !synonymNames.isEmpty()) {
            List<String> quotedFuncNames = new ArrayList<>();
            for (String synonymName: synonymNames) {
                quotedFuncNames.add("'" + synonymName + "'");
            }
            sql += String.format(" WHERE SYNONYM_NAME IN (%s) and owner = '%s'", String.join(",", quotedFuncNames), schema);
        }
        List<UserDefSynonym> userDefSynonyms = new ArrayList<>();
        String dbUrl = defaultUrl;
        Connection connection = getConnection(dbUrl);

        try (Statement stmt = connection.createStatement()){
            try (ResultSet rs = stmt.executeQuery(sql)){
                while (rs.next()) {
                    UserDefSynonym userDefSynonym = new UserDefSynonym();
                    String synonymName = rs.getString("SYNONYM_NAME");
                    userDefSynonym.setSynonymName(synonymName);
                    String ddlQuery = String.format("SELECT DBMS_METADATA.GET_DDL('%s', '%s', '%s') AS DDL FROM dual",
                            "SYNONYM",
                            synonymName,
                            schema);
                    try (PreparedStatement ps = connection.prepareStatement(ddlQuery)){
                        ResultSet ddlResultSet = ps.executeQuery();
                        if (ddlResultSet.next()) {
                            userDefSynonym.setDdl(ddlResultSet.getString("DDL"));
                        }
                    } catch (SQLException e) {
                        e.printStackTrace();
                        throw new CatalogException(APIErrorCode.QUERY_FUNCTION_DDL_ERROR, "查询自定义同义词的DDL失败：synonym=" + userDefSynonym, e);
                    }
                    userDefSynonyms.add(userDefSynonym);
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.QUERY_FUNCTION_LIST_ERROR, "查询所有自定义同义词的DDL失败，请检查SQL语句：sql=" + sql, e);
        }
        return userDefSynonyms;
    }

    /**
     * 对Oracle，已经指定模式了，查询表无需前缀
     * @param tablePath
     * @return
     */

    protected String getSelectColumnsSql(TablePath tablePath) {
        return "SELECT * FROM " + tablePath.getSchemaAndTableName();
    }

    public String getUrlFromDatabaseName(String database) {
        return defaultUrl;
    }

    @Override
    public List<String> specialColumns() {
        return List.of("XMLTYPE", "XMLINDEX", "RAW");
    }

    @Override
    public List<String> bigDataColumns() {
        return List.of("LONG", "LONG RAW", "BLOB", "CLOB", "NCLOB", "BFILE");
    }

    @Override
    protected String getColumnsSql(TablePath tablePath) {
        String sql = "SELECT a.owner as catalog_name, a.owner as schema_name, a.table_name, a.column_name, a.data_type as column_type " +
                " , (case \n" +
                "                            when a.data_type = 'NUMBER' \n" +
                "                            then (CASE\n" +
                "\tWHEN a.data_precision IS NULL THEN\n" +
                "\t\t0\n" +
                "\tELSE\n" +
                "\t\ta.data_precision end\n" +
                ")     \n" +
                "                            when a.data_type in ('CHAR','VARCHAR2','NVARCHAR2')\n" +
                "                            then char_col_decl_length\n" +
                "                            when LOWER(a.data_type) in ('binary_double','binary_float','blob','clob','long',\n" +
                "                            'long raw','nclob','raw','date')\n" +
                "                            then data_length\n" +
                "                            else 0\n" +
                "                            end\n" +
                "                        ) as data_length  " +
                " , CASE  " +
                "  WHEN a.nullable = 'Y' THEN 1 " +
                "  ELSE 0 " +
                " END AS nullable, a.data_default, b.comments as column_comment " +
                " , CASE  " +
                "  WHEN t.column_name IS NOT NULL THEN 1 " +
                "  ELSE 0 " +
                " END AS is_primary_key " +
                "FROM all_tab_columns a " +
                " INNER JOIN all_col_comments b " +
                " ON a.owner = b.owner " +
                "  AND a.table_name = b.table_name " +
                "  AND a.column_name = b.column_name " +
                " LEFT JOIN ( " +
                "  SELECT cu.column_name AS column_name " +
                "  FROM all_cons_columns cu, all_constraints au " +
                "  WHERE cu.owner = upper('" + tablePath.getSchemaName() + "') " +
                "   AND cu.owner = au.owner " +
                "   AND cu.constraint_name = au.constraint_name " +
                "   AND au.constraint_type = 'P' ";
        if (StringUtils.isNotEmpty(tablePath.getTableName())) {
            sql = sql + "   AND au.table_name in ('" + tablePath.getTableName() + "') ";
        }
        sql = sql + " ) t " +
                " ON a.column_name = t.column_name " +
                "WHERE a.owner = upper('" + tablePath.getSchemaName() + "') ";
        if (StringUtils.isNotEmpty(tablePath.getTableName())) {
            sql = sql + " AND a.Table_Name in ('" + tablePath.getTableName() + "') ";
        }
        sql = sql + "ORDER BY column_id ASC";
        return sql;
    }

    @Override
    public List<String> findFunctionIncompatible(String sql, String sourceCatalog, String sinkCatalog) {
        CharStream input = CharStreams.fromString(sql);

        // 创建词法分析器和解析器
        PlSqlLexer lexer = new PlSqlLexer(input);
        CommonTokenStream tokens = new CommonTokenStream(lexer);
        PlSqlParser parser = new PlSqlParser(tokens);

        // 从解析器中获取起始规则并进行解析
        ParseTree tree = parser.sql_script();

        // 创建ParseTreeWalker
        ParseTreeWalker walker = new ParseTreeWalker();

        // 使用自定义Listener遍历解析树
        Oracle2DMFunctionListener listener = new Oracle2DMFunctionListener(sourceCatalog, sinkCatalog);
        walker.walk(listener, tree);
        return listener.getDifferentFunctions();
    }

    @Override
    public List<String> listTypeColumns(String owner, String typeName) throws CatalogException {
        String trimedTypeName = StringTool.removeDoubleQuotes(typeName);
        String sql = String.format("SELECT ATTR_NAME FROM ALL_TYPE_ATTRS WHERE OWNER = '%s' AND TYPE_NAME = '%s'  ORDER BY ATTR_NO ", owner, trimedTypeName);
        return listNames(sql, "ATTR_NAME");
    }

    @Override
    protected String getDataSqlNoPk(TablePath tablePath, int limit, int offset) {
        return String.format("SELECT *\n" +
                "FROM (SELECT B.*, ROWNUM RN2\n" +
                "      FROM (SELECT ROWNUM RN1, A.*\n" +
                "            FROM %s A) B\n" +
                "      ORDER BY RN1 DESC)\n" +
                "WHERE ROWNUM < %s\n", getFullTableName(tablePath), limit + 1);
    }

    @Override
    public void setCurrentSchema(Connection connection, String schema) throws SQLException {
        String sql = "ALTER SESSION SET CURRENT_SCHEMA = " + schema;
        connection.createStatement().execute(sql);
    }

    @Override
    protected String getRandomPksSql(TablePath tablePath, String pk, int limit, int offset) {
        return String.format("SELECT %s \n" +
                "FROM (\n" +
                "    SELECT e.*, ROW_NUMBER() OVER (ORDER BY %s DESC) AS rn\n" +
                "    FROM %s e\n" +
                ") \n" +
                "WHERE rn <= %s\n", pk, pk, getFullTableName(tablePath), limit);
    }

    @Override
    public String getCreateTableSql(TablePath tablePath, CatalogTable table, TaskConf taskConf) throws DataTypeMappingNotExistException {
        return super.getCreateTableSql(tablePath, table, taskConf);
    }

    @Override
    public String[] getScriptDelimiters() {
        return SQLConstants.DEFAULT_SCRIPT_DELIMITER; //$NON-NLS-1$
    }

    @Override
    public String[][] getBlockBoundStrings() {
        return new String[][]{
                {SQLConstants.BLOCK_BEGIN, SQLConstants.BLOCK_END},
                {"IF", SQLConstants.BLOCK_END},
                {"LOOP", SQLConstants.BLOCK_END + " LOOP"},
                {SQLConstants.KEYWORD_CASE, SQLConstants.BLOCK_END + " " + SQLConstants.KEYWORD_CASE},
        };
    }
}
